package com.atguigu.gmall.realtime.app.functions;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * Author: tiancy
 * Date: 2021/11/26
 * Desc: 业务数据动态分流实现类
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    //定义的侧输出流标签,将标签传进来.
    private OutputTag<JSONObject> dimTag;
    // 定义当前合并流中的 广播状态(FlinkCDC中获取的数据会存成状态并广播进来)
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunction(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        //创建连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    // 处理主流数据业务
    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> out) throws Exception {
        // 先来获取广播流中的状态信息.是一个map<表名:操作,当前表的详细信息对象>
        ReadOnlyBroadcastState<String, TableProcess> tableProcessInfo = readOnlyContext.getBroadcastState(mapStateDescriptor);
        //获取当前处理行数据的表名
        String curJsonObjTableName = jsonObject.getString("table");
        //获取操作类型
        String curJsonObjType = jsonObject.getString("type");

        //注意：后续我们使用maxwell提供的对历史数据进行处理的命令maxwell-bootstrap对历史数据进行处理的时候，类型为bootstrap-insert
        if ("bootstrap-insert".equals(curJsonObjType)) {
            curJsonObjType = "insert";
            jsonObject.put("type", curJsonObjType);
        }

        //根据表名和操作类型拼接key
        String key = curJsonObjTableName + ":" + curJsonObjType;

        if (!"table_process".equals(curJsonObjTableName)) {  // 这里直接指定 maxwall 不需要监测 配置表 table_process中的变化信息. -->直接使用的是FlinkCDC来监测的.
            // 通过组合的key去状态中查找当前处理行所属表的具体信息.
            TableProcess tableProcess = tableProcessInfo.get(key);
            if (tableProcess != null) {
                // 通过tableProcess对象确定数据去向, 是Kafka中作为事实表数据还是发往HBase作为维度表数据.
                String sinkType = tableProcess.getSinkType();
                // 最重要的一点 : 无论数据最终去向为哪里,都需要加入最终发往目的地的信息,比如 Kafka中具体的主题,或者HBase中具体的表中.
                jsonObject.put("sink_table", tableProcess.getSinkTable());

                /*
                    这里需要对 maxwall检测到的变化数据,进行格式上的处理,就是我在往下游 Kafka或者HBase中存的时候,只需要存 配置表 table_process中指定的字段所对应的值即可,不需要maxwall检测到的其他属性的数据.
                    maxwall检测到的变化数据格式如下 :
                    ====dim层数据====:3>
                    {
                      "database":"gmall0609",
                      "xid":117349,
                      "data":{"tm_name":"巴黎世家hs","logo_url":"yyds","id":14},
                      "commit":true,
                      "sink_table":"dim_base_trademark",
                      "type":"insert",
                      "table":"base_trademark",
                      "ts":1638106931
                    }


                    >>>>>dwd层数据:<<<<<:4>
                    {
                      "database":"gmall0609",
                      "xid":117382,
                      "data":{"delivery_address":"第18大街第6号楼4单元197门","order_comment":"描述933556","original_total_amount":27591.00,"order_status":"1005","consignee_tel":"13883959408","trade_body":"Apple iPhone 12 (A2404) 128GB 黑色 支持移动联通电信5G 双卡双待手机等3件商品","id":29245,"operate_time":"2021-11-26 22:15:56","consignee":"司马老贼","create_time":"2021-11-26 22:15:55","expire_time":"2021-11-26 22:30:55","coupon_reduce_amount":0.00,"out_trade_no":"295391219485323","total_amount":27597.00,"user_id":184,"img_url":"http://img.gmall.com/741356.jpg","province_id":22,"feight_fee":6.00,"activity_reduce_amount":0.00},
                      "old":{"consignee":"司马祥才"},
                      "commit":true,
                      "sink_table":"dwd_order_info",
                      "type":"update",
                      "table":"order_info",
                      "ts":1638106951
                    }

                 */
                // maxwall检测到变化的数据`data属性对应的值` 进行一个过滤,筛选出我们想要的属性以及值.
                JSONObject dataJsonObj = jsonObject.getJSONObject("data");
                filterColumn(dataJsonObj, tableProcess.getSinkColumns());
                if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) { // 如果是发往HBase的数据,直接写到侧输出流中.
                    readOnlyContext.output(dimTag, jsonObject);
                } else { // 如果不是维度数据,就是事实数据,因该直接留在主流中,将来发往 Kafka中指定的分区.
                    out.collect(jsonObject);
                }
            } else {
                System.out.println("当前数据中所属的表以及指定的操作符: " + key + "在配置表 gmall0609_realtime.table_process 未配置");
            }
        }

    }

    /**
     * 对maxwall检测到的数据进行格式上的过滤,再发往下游算子进行写入.
     *
     * @param dataJsonObj  当前行数据中的`data`属性 的转化成的json对象
     * @param sinkColumns 当前数据行中,所对应的表结构.
     */
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        //将保留的字段放到集合中  方便进行包含判断
        String[] fieldArr = sinkColumns.split(",");
        List<String> fieldList = Arrays.asList(fieldArr);
        //获取json对象的所有名值对
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        // 这里注意,在删除集合中的元素时,不能直接通过遍历删除,而是要使用迭代器对象中的remove()方法来删除. 下面这种写法,是java8提供的一种lambada表达式写法.
        entrySet.removeIf(entry -> !fieldList.contains(entry.getKey()));
    }

    // 处理侧输出流中的业务.
    @Override
    public void processBroadcastElement(String input, Context context, Collector<JSONObject> collector) throws Exception {
        //jsonStr  是FlinkCDC从MySQL中读取配置信息，然后我们通过自定义反序列化器的方式  将其封装为jsonStr
        /* TODO 获取当前`FlinkCDC`中监测到的配置表中修改变化的数据,并将变化的数据经过数据组合,设置成当前
                        {
                            "database":"gmall0609_realtime",
                            "data":{"operate_type":"insert","sink_type":"hbase",sink_table":"dim_base_trademark","source_table":"base_trademark","sink_pk":"id","sink_columns":"id,tm_name"},
                            "type":"insert",
                            "table":"table_process"
                        }
                    */
        //为了处理方便，将json字符串转换为json对象
        JSONObject jsonObj = JSON.parseObject(input);
        //获取从配置表中读取的一条配置记录--->我们可以将其转换为TableProcess实体类对象
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");
        /*
                    将配置记录封装为TableProcess对象.
                    这里将json对象转化为实体类时,会那当前的json对象中的属性,也就是像 operate_type、sink_type、sink_table、source_table、sink_pk、sink_columns、sink_extend等字段来与
                    实体类中属性进行匹配[sourceTable、operateType、sinkType、sinkColumns、sinkPk、sinkExtend] 进行匹配,匹配规则 : 数据库中的下划线可以转化为小驼峰命名再进行匹配.
                 */
        TableProcess tableProcess = dataJsonObj.toJavaObject(TableProcess.class);
        //获取原始数据库表名
        String sourceTable = tableProcess.getSourceTable();
        //获取操作类型
        String operateType = tableProcess.getOperateType();
        //获取是维度还是事实
        String sinkType = tableProcess.getSinkType();
        //获取输出的目的地
        String sinkTable = tableProcess.getSinkTable();
        //获取建表字段
        String sinkColumns = tableProcess.getSinkColumns();
        //获取建表扩展
        String sinkExtend = tableProcess.getSinkExtend();
        //获取主键
        String sinkPk = tableProcess.getSinkPk();

        //根据表名和操作类型 拼接key,key的格式 如 dim_base_trademark:insert,value中的数据一个TableProcess对象.
        String key = sourceTable + ":" + operateType;
        /*
            如果是维度数据，需要通过Phoenix创建表
            对读取到的配置信息进行判断     如果是维度配置，并且配置的是insert操作，那么提前创建维度表
         */
        if (TableProcess.SINK_TYPE_HBASE.equals(sinkType) && "insert".equals(operateType)) {
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
        }
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        //将配置信息放到状态中保存
        broadcastState.put(key, tableProcess);
    }

    /**
     * 如果是维度数据,需要通过phoenix连接HBase判断表是否存在.如果不存在,则需要创建表.
     *
     * @param sinkTable   : 判断获取的 配置表中的属性,对应下游发往的位置 HBase or Kafka
     * @param sinkColumns : 表中的字段名称
     * @param sinkPk      : 表的主键
     * @param sinkExtend  : 建表拓展,建表时是否指定存储引擎,是否指定表中的编码格式.
     */
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        //处理主键或者建表扩展为null的情况
        if (sinkPk == null) {
            sinkPk = "id";
        }
        if (sinkExtend == null) {
            sinkExtend = "";
        }
        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists " + GmallConfig.HBASE_SCHEMA + "." + sinkTable + "(");

        //建表字段的处理
        String[] fieldArr = sinkColumns.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String field = fieldArr[i];
            if (i >= 1) {
                createSql.append(",");
            }
            if (field.equals(sinkPk)) {
                createSql.append(field).append(" varchar primary key ");
            } else {
                createSql.append(field).append(" varchar ");
            }

           /* //如果是最后一个字段，那不需要加逗号
            if(i < fieldArr.length -1){
                createSql.append(",");
            }*/
        }

        createSql.append(")").append(sinkExtend);

        System.out.println("在phoenix中建表的语句为:" + createSql);
        PreparedStatement ps = null;
        try {
            //获取数据库操作对象
            ps = conn.prepareStatement(createSql.toString());
            //执行SQL语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("在phoenix中创建维度表失败");
        } finally {
            //释放资源
            if (ps != null) {
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
